package com.example.canal.app.kafka;

import com.example.canal.app.ActionByEventType;
import com.example.canal.conf.CanalConstants;
import com.example.canal.conf.KafkaSyncProperties;
import com.example.canal.event.EventInfo;
import com.example.canal.service.IEventHandler;
import com.example.canal.service.kafka.KafkaHandler;
import com.google.common.base.Splitter;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

@Component(value = "kafkaSyncApp")
public class KafkaSyncApp implements ActionByEventType {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSyncApp.class);

    @Autowired
    private KafkaSyncProperties kafkaSyncProperties;

    @Autowired
    private KafkaHandler kafkaHandler;

    @Override
    public IEventHandler createHandlerByEventType(EventInfo eventInfo) {
        if (Objects.isNull(eventInfo)) {
            LOGGER.info("创建kafka处理器失败,eventInfo为空");
            return null;
        }
        if (Objects.equals(kafkaSyncProperties.isEnable(), Boolean.TRUE)) {
            LOGGER.info("kafkaSyncProperties={}",kafkaSyncProperties);
            String keys = this.kafkaSyncProperties.getKeys();
            if (StringUtils.isNotBlank(keys)) {
                if (!Objects.equals(keys, CanalConstants.DEFAULT_FILTER)) {
                    List<String> keyArr = Splitter.on(",").splitToList(keys);
                    if (keyArr.contains(eventInfo.getTableInfo().getSchemaName() + "." +
                            eventInfo.getTableInfo().getTableName())) {
                        return kafkaHandler;
                    }
                }
            }
            return kafkaHandler;
        }
        return null;
    }


}
